/**
 * Project: shuyin-sync
 * 
 * File Created at Mar 20, 2012
 * $Id$
 * 
 * Copyright 2008 6677bank.com Croporation Limited.
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of
 * 6677bank Company. ("Confidential Information").  You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with 6677bank.com.
 */
package com.bank6677.sync.csm.service.impl;

import java.sql.SQLException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.ListUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import com.bank6677.application.Constants;
import com.bank6677.common.exception.BusinessException;
import com.bank6677.common.exception.SystemException;
import com.bank6677.sync.csm.ApplicationSyncExecutor;
import com.bank6677.sync.csm.dao.ApplicationDao;
import com.bank6677.sync.csm.dao.ClassificationDao;
import com.bank6677.sync.csm.domain.Application;
import com.bank6677.sync.csm.domain.ApplicationExt;
import com.bank6677.sync.csm.domain.ApplicationStatFlow;
import com.bank6677.sync.csm.domain.Classification;
import com.bank6677.sync.csm.domain.DropApplication;
import com.bank6677.sync.csm.domain.Reservation;
import com.bank6677.sync.csm.domain.SendInstitutionFlow;
import com.bank6677.sync.csm.domain.UrgeApp;
import com.bank6677.sync.csm.helper.ApplicationHelper;
import com.bank6677.sync.csm.helper.ClassificationHelper;
import com.bank6677.sync.csm.service.ApplicationSyncRMIService;
import com.bank6677.sync.csm.service.ApplicationSyncService;

/**
 * 订单同步相关接口实现类
 * 
 * @author liyue
 * @version ApplicationSyncServiceImpl.java Mar 20, 2012 2:34:44 PM
 */
@Service
@Scope("prototype")
public class ApplicationSyncServiceImpl implements ApplicationSyncService, ApplicationSyncRMIService {

    static final Logger                LOG             = Logger.getLogger(ApplicationSyncServiceImpl.class);

    @Autowired
    private ApplicationDao             applicationDao;
    @Autowired
    private ClassificationDao          classificationDao;
    @Autowired
    @Qualifier("cmbankTransactionManager")
    private PlatformTransactionManager cmbankTransactionManager;

    private Integer                    totalAppNum     = 0;                                                 //需要同步的总数
    private Integer                    currentStartApp = 0;                                                 //每次循环开始的基数值
    private Integer                    peroSyncAppNum  = 100;                                               //每次批量插入的量

    private static final AtomicInteger syncCount       = new AtomicInteger(0);
    private static final AtomicInteger totalCount      = new AtomicInteger(0);

    public int countApplications() throws SQLException {
        return applicationDao.countOldApplication(new HashMap<String, Object>());
    }

    @Override
    @Transactional(value = "cmbankTransactionManager", propagation = Propagation.NEVER)
    public void syncApplications() throws SQLException {

        totalAppNum = applicationDao.countOldApplication(new HashMap<String, Object>());

        for (; currentStartApp < totalAppNum;) {

            LOG.debug("Start synchronize applications from newpayx, limit " + currentStartApp + ", " + peroSyncAppNum);
            final List<Application> apps = applicationDao.findOldApplication2new(currentStartApp,
                    currentStartApp + peroSyncAppNum > totalAppNum ? totalAppNum - currentStartApp : peroSyncAppNum);

            try {
                syncApplicationsWithTransaction(apps);
            } catch (BusinessException e) {
                LOG.error(e.getMessage(), e);
                continue;
            } finally {
                currentStartApp += peroSyncAppNum;
            }
        }
    }

    public void syncApplications(List<Long> ids) throws SQLException {

        if (CollectionUtils.isNotEmpty(ids)) {
            totalAppNum = ids.size();
        } else {
            throw new BusinessException("请输入需要同步的单号串！");
        }

        LOG.info("Start synchronize applications by list of id. [" + StringUtils.join(ids, ",") + "]");

        for (; currentStartApp < totalAppNum;) {
            try {
                List<Application> apps = applicationDao.findOldApplication2new(ids);
                syncApplicationsWithTransaction(apps);
            } catch (Exception e) {
                LOG.error(e.getMessage(), e);
                continue;
            } finally {
                currentStartApp += peroSyncAppNum;
            }
        }
    }

    public void syncApplications(String from, String to) throws SQLException {

        Map<String, String> conditions = new HashMap<String, String>();
        conditions.put("from", from);
        conditions.put("to", to);
        totalAppNum = applicationDao.countOldApplication(conditions);

        for (; currentStartApp < totalAppNum;) {

            LOG.debug("Start synchronize applications from newpayx, limit " + currentStartApp + ", " + peroSyncAppNum);
            List<Application> apps = applicationDao.findOldApplication2new(currentStartApp,
                    currentStartApp + peroSyncAppNum > totalAppNum ? totalAppNum - currentStartApp : peroSyncAppNum, from, to);

            try {
                syncApplicationsWithTransaction(apps);
            } catch (BusinessException e) {
                LOG.error(e.getMessage(), e);
                continue;
            } finally {
                currentStartApp += peroSyncAppNum;
            }
        }
    }

    public void asyncApps() throws SQLException {

        final ThreadPoolExecutor executorPool = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(5),
                new ThreadPoolExecutor.CallerRunsPolicy());
        //        executorPool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());

        //        final ExecutorService executorPool = Executors.newFixedThreadPool(10);
        totalAppNum = applicationDao.countOldApplication(new HashMap<String, Object>());

        if (totalAppNum >= 1000) {
            for (; currentStartApp < totalAppNum;) {
                List<Application> apps = applicationDao.findOldApplication2new(currentStartApp, currentStartApp + 1000 > totalAppNum ? totalAppNum
                        - currentStartApp : 1000);

                executorPool.execute(new ApplicationSyncExecutor(apps));
                currentStartApp += 1000;
            }
        } else {
            List<Application> apps = applicationDao.findOldApplication2new(currentStartApp, currentStartApp + 1000 > totalAppNum ? totalAppNum
                    - currentStartApp : 1000);
            executorPool.execute(new ApplicationSyncExecutor(apps));
        }

        executorPool.shutdown();
    }

    public void syncApplicationsWithTransaction(List<Application> apps) {

        DefaultTransactionDefinition definition = new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
        TransactionStatus status = cmbankTransactionManager.getTransaction(definition);

        long start = System.currentTimeMillis();

        List<Application> syncedApps = new ArrayList<Application>();

        for (Application application : apps) {
            try {
                if (!ApplicationHelper.checkOldTypeAndPurpose(application.getLoanType(), application.getLoanPurpose())) {
                    continue;//如果这条订单的类型和用途无法匹配，则不同步
                }

                if (ApplicationHelper.checkLinkMan(application.getLinkMan())) {
                    continue;//
                }

                if (ApplicationHelper.checkPhone(application.getLinkPhone())) {
                    continue;//
                }

                syncSingleApplication(application);
                syncedApps.add(application);
            } catch (Exception err) {
                cmbankTransactionManager.rollback(status);

                throw new BusinessException("Executor stopped where " + currentStartApp + "-" + (currentStartApp + peroSyncAppNum)
                        + ". Error occured in synchronize applications, id:" + application.getApplyId(), err);
            }
            LOG.debug(currentStartApp + "-" + (currentStartApp + peroSyncAppNum > totalAppNum ? totalAppNum : currentStartApp + peroSyncAppNum)
                    + ", id:" + application.getApplyId() + " done.");
        }

        cmbankTransactionManager.commit(status);
        syncCount.addAndGet(CollectionUtils.size(syncedApps));
        totalCount.addAndGet(CollectionUtils.size(apps));
        LOG.error("sync:" + syncCount + ". total:" + totalCount + "[" + StringUtils.join(ListUtils.removeAll(apps, syncedApps), ",")
                + "] were failed. ");
        LOG.info(syncCount + "[" + StringUtils.join(syncedApps, ",") + "] were succeed. Use time " + (System.currentTimeMillis() - start));
    }

    /**
     * 同步某笔订单
     * 
     * @param application
     * @throws SQLException
     */
    private void syncSingleApplication(Application application) throws SQLException {
        Integer loanType = application.getLoanType();
        Integer loanPurpose = application.getLoanPurpose();

        ///// tableName, modelIds /////
        application.setTableName(ApplicationHelper.getNewTableName(loanType, loanPurpose));
        application.setModelIds(ApplicationHelper.getNewModelIds(loanType, loanPurpose));

        /////   subject, type, usage    /////
        application.setLoanType(ApplicationHelper.getNewLoanType(loanType, loanPurpose));
        application.setLoanPurpose(ApplicationHelper.getNewLoanPurpose(loanType, loanPurpose));
        application.setSubject(ApplicationHelper.getNewLoanSubject(loanType, loanPurpose));

        ///// 大小状态更新  /////
        application.setProStat(ApplicationHelper.getNewProStatus(application.getProcessStat()));
        application.setProcessStat(ApplicationHelper.getNewProcessStatus(application.getProcessStat()));
        application.setLastProStat(ApplicationHelper.getNewProcessStatus(application.getProcessStat()));

        ///// 订单流转类型   /////
        application.setAppType(Constants.APP_TYPE_MOVE);

        ///// 该订单的催缴  /////
        application.setUrges(getNewUrges(application));

        ///// 该订单的操作日志  /////
        application.setFlows(getNewFlows(application));

        ///// 订单的弃单记录    /////
        application.setDrops(getDrops(application));

        ///// 订单的预约记录    /////
        application.setReservations(getReservations(application));

        ////  订单机构派发记录  ////
        application.setSendFlow(getSendFlows(application));

        ////  订单附加信息记录      ////
        application.setExt(getAppExt(application));

        applicationDao.saveApplication(application);
    }

    /**
     * <p>
     * Returns reservations of single application from old version.
     * 
     * @param application
     * @return
     */
    private List<SendInstitutionFlow> getSendFlows(Application application) {
        List<SendInstitutionFlow> sendFlows = null;
        try {
            sendFlows = applicationDao.findOldSendFlowByAppId(application.getApplyId());
            for (SendInstitutionFlow flow : sendFlows) {
                //TODO 机构ID及机构名
                flow.setState(ApplicationHelper.getNewProcessStatus(flow.getState()));

                if (flow.getOperAdminId() == null) {
                    flow.setOperAdminId(application.getAdminId());
                }
            }
        } catch (SQLException e) {
            throw new SystemException("Error occured when query send institution flows, id:" + application.getApplyId(), e);
        }
        return sendFlows;
    }

    /**
     * <p>
     * Returns reservations of single application from old version.
     * 
     * @param application
     * @return
     */
    private List<Reservation> getReservations(Application application) {
        List<Reservation> reservations = null;
        try {
            reservations = applicationDao.findOldReservationByAppId(application.getApplyId());
            if (CollectionUtils.isEmpty(reservations)) {
                for (Reservation reserve : reservations) {
                    reserve.setResType(ClassificationHelper.getNewReservationType(reserve.getResType()));
                    reserve.setResStatus(ApplicationHelper.getNewReservationStatus(reserve.getResStatus()));
                    /// by liyue  change res_remark varchar2(100) to varchar2(512)

                    /////如果订单的创建日期为空，就将其订单的创建日期根据订单号来创建
                    if (reserve.getCreateDt() == null) {
                        try {
                            reserve.setCreateDt(DateUtils.parseDate(StringUtils.substring(reserve.getApplyId().toString(), 0, 13),
                                    new String[] { "yyyyMMddHHmmss" }));
                        } catch (ParseException e) {
                            throw new BusinessException("日期转换出错！！", e);
                        }
                    }
                }
            }
        } catch (SQLException e) {
            throw new SystemException("Error occured when query reservations, id:" + application.getApplyId(), e);
        }
        return reservations;
    }

    /**
     * <p>
     * Returns drop logs of single application from old version.
     * 
     * @param application
     * @return
     */
    private List<DropApplication> getDrops(Application application) {
        List<DropApplication> drops = null;
        try {
            drops = applicationDao.findOldDropFlowByAppId(application.getApplyId());
        } catch (SQLException e) {
            throw new SystemException("Error occured when query drop logs, id:" + application.getApplyId(), e);
        }
        return drops;
    }

    /**
     * <p>
     * Returns urges of single application from old version.
     * 
     * @param application
     * @return
     */
    private List<UrgeApp> getNewUrges(Application application) {
        List<UrgeApp> urges = null;
        try {
            urges = applicationDao.findOldUrgeByApplyId(application.getApplyId());
        } catch (SQLException e) {
            throw new SystemException("Error occured when query urges, id:" + application.getApplyId(), e);
        }
        return urges;
    }

    /**
     * <p>
     * Returns operation logs of single application from old version, then
     * package to new version.
     * 
     * @param application
     * @return
     */
    private List<ApplicationStatFlow> getNewFlows(Application application) {
        List<ApplicationStatFlow> flows = null;
        try {
            flows = applicationDao.findOldOptLogsByApplyId(application.getApplyId());
            if (!CollectionUtils.isEmpty(flows)) {
                for (ApplicationStatFlow flow : flows) {
                    flow.setOperationId(ApplicationHelper.getNewOperation(flow.getOperationId()));
                    flow.setLogType(new Integer(flow.getLogBelong()));
                }
            }
        } catch (SQLException e) {
            throw new SystemException("Error occured when query operation logs, id:" + application.getApplyId(), e);
        }
        return flows;
    }

    /**
     * <p>
     * Returns extension information of single application from old version,
     * then package to new version.
     * 
     * @param application
     * @return
     */
    private ApplicationExt getAppExt(Application application) {
        ApplicationExt ext = null;
        try {
            ext = applicationDao.findAppExtsByAppId(application.getApplyId());
            ext.setAdNameStr(ApplicationHelper.getPropertyValueInAppExt("adName", ext.getAdNameStr()));
            ext.setAdTypeStr(ApplicationHelper.getPropertyValueInAppExt("adType", ext.getAdTypeStr()));

            ext.setAdName("false".equals(ext.getAdNameStr()) ? null : NumberUtils.createInteger(ext.getAdNameStr()));
            ext.setAdType("false".equals(ext.getAdTypeStr()) ? null : NumberUtils.createInteger(ext.getAdTypeStr()));
            ext.setUserSex(ApplicationHelper.getPropertyValueInAppExt("userSex", ext.getUserSex()));
            ext.setXyzk(ApplicationHelper.getPropertyValueInAppExt("xyzk", ext.getXyzk()));
            ext.setHyzk(ApplicationHelper.getPropertyValueInAppExt("hyzk", ext.getHyzk()));
            ext.setWorkActor(ApplicationHelper.getPropertyValueInAppExt("workActor", ext.getWorkActor()));
            ext.setWorkZw(ApplicationHelper.getPropertyValueInAppExt("workZw", ext.getWorkZw()));
            ext.setJycd(ApplicationHelper.getPropertyValueInAppExt("jycd", ext.getJycd()));
            ext.setSalaryType(ApplicationHelper.getPropertyValueInAppExt("salaryType", ext.getSalaryType()));
            ext.setWorkDwxz(ApplicationHelper.getPropertyValueInAppExt("workDwxz", ext.getWorkDwxz()));
        } catch (SQLException e) {
            throw new SystemException("Error occured when query extension information of application, id:" + application.getApplyId(), e);
        }

        return ext;
    }

    @Override
    @Transactional(value = "cmbankTransactionManager")
    public void syncRejectType() {
        try {
            List<Classification> cfns = classificationDao.findAllRejectTypes2new();
            for (Classification cfn : cfns) {
                cfn.setActive(ClassificationHelper.getNewStatus(cfn.getActive()));
                cfn.setCategoryType(ClassificationHelper.CATEGORY_DROP);
                cfn.setCategoryId(ClassificationHelper.PRE_OLD_REJECT_CATEGORY_ID + cfn.getCategoryId());
                cfn.setCategoryParentId(cfn.getCategoryParentId() == 0 ? 0 : ClassificationHelper.PRE_OLD_REJECT_CATEGORY_ID
                        + cfn.getCategoryParentId());
            }

            classificationDao.batchCreateNewClassification(cfns);
        } catch (SQLException err) {
            throw new BusinessException(err.getMessage(), err);
        }
    }

}
